package com.gmall.data.functions;

import com.gmall.data.api.UserActionCountQueryService;
import com.gmall.data.api.UserActionSequenceQueryService;
import com.gmall.data.api.UserProfileQueryService;
import com.gmall.data.pojo.LogBean;
import com.gmall.data.pojo.ResultBean;
import com.gmall.data.pojo.RuleParam;
import com.gmall.data.service.UserActionCountQueryServiceImpl;
import com.gmall.data.service.UserActionSequenceQueryServiceImpl;
import com.gmall.data.service.UserProfileQueryServiceImpl;
import com.gmall.data.utils.RuleSimulator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 规则核心处理函数
 */
public class RuleProcessFunction extends KeyedProcessFunction<String, LogBean, ResultBean> {

    private UserProfileQueryService userProfileQueryService;
    private UserActionCountQueryService userActionCountQueryService;
    private UserActionSequenceQueryService userActionSequenceQueryService;

    ListState<LogBean> eventState;

    RuleParam ruleParam;

    @Override
    public void open(Configuration parameters) throws Exception {

        /**
         * 准备一个存储明细事件的state，存储该用户上线以来的所有行为明细数据
         */
        ListStateDescriptor<LogBean> eventStateDesc = new ListStateDescriptor<>("eventState", LogBean.class);
        eventState = getRuntimeContext().getListState(eventStateDesc);

        /**
         * 构造底层的核心查询服务
         */
        userProfileQueryService     = new UserProfileQueryServiceImpl();
        userActionCountQueryService = new UserActionCountQueryServiceImpl(eventState);
        userActionSequenceQueryService = new UserActionSequenceQueryServiceImpl(eventState);

        /**
         * 获取规则参数
         */
        ruleParam = RuleSimulator.getRuleParam();

    }

    @Override
    public void processElement(LogBean logBean, Context context, Collector<ResultBean> out) throws Exception {
        eventState.add(logBean);

        // 判断是否满足触发条件
        if (ruleParam.getTriggerParam().getEventId().equals(logBean.getEventId())) {

            // 查询画像条件
            boolean isProfile = userProfileQueryService.judgeProfileCondition(logBean.getDeviceId(), ruleParam);
            if (!isProfile) return;

            // 查询行为次序条件
            boolean isCount = userActionCountQueryService.queryActionCounts(logBean.getDeviceId(), ruleParam);
            if (!isCount) return;

            // 查询用户序列条件
            boolean isSeq = userActionSequenceQueryService.queryActionSequence(logBean.getDeviceId(), ruleParam);
            if (!isSeq) return;;


            // 输出一个规则匹配成功的结果
            ResultBean resultBean = new ResultBean();
            resultBean.setRuleId(ruleParam.getRuleId());
            resultBean.setDeviceId(logBean.getDeviceId());
            resultBean.setTimeStamp(logBean.getTimeStamp());

            out.collect(resultBean);

        }


    }
}
